import json
import re
import time

from kafka import KafkaConsumer, KafkaProducer
from config.config import SCHEMA_CHANGE_KAFKA
from tool import send_email
from tool.MysqlHelper import MysqlHelper
from tool.logger import logger


class DorisSchemaChange:
    def __init__(self):
        self.consumer = None
        self.config = SCHEMA_CHANGE_KAFKA
        self.bootstrap_servers = self.config['bootstrap_servers'].split(",")
        self.auto_offset_reset = self.config['auto_offset_reset']
        self.topic = self.config['topic']
        self.group_id = self.config['group_id']
        # self.group_id = 'local_test32'
        self.get_consumer()
        self.producer = KafkaProducer(bootstrap_servers=self.config['bootstrap_servers'])
        self.error_topic = "errors"
        self.sql = []
        self.last_table = ''
        self.last_timestamp = 0
        self.error_msg = "项目:DorisSchemaChange\n模块:{}\n库:{}\n表{}\n时间:{}"

    def get_consumer(self):
        self.consumer = KafkaConsumer(
            self.topic,
            bootstrap_servers=self.bootstrap_servers,
            auto_offset_reset=self.auto_offset_reset,
            value_deserializer=lambda m: json.loads(m.decode("utf-8")),
            enable_auto_commit=True,
            group_id=self.group_id,

        )

    def get_columns(self,TABLE_SCHEMA,TABLE_NAME):
        sql=f"select COLUMN_NAME,DATA_TYPE from information_schema.columns where TABLE_SCHEMA='{TABLE_SCHEMA}' and TABLE_NAME='{TABLE_NAME}'"
        doris_db = MysqlHelper('information_schema')
        result=doris_db.select_with_schema(sql)
        res={}
        for column in result:
            res[column['COLUMN_NAME']]=str(column['DATA_TYPE']).upper()
        return res

    def change(self):
        for info in self.consumer:

            data = info.value['source']
            db = data['db']
            table = data['table']
            historyRecord = json.loads(info.value['historyRecord'])
            logger.info(f'db:{db},table:{table}')
            ddl = str(historyRecord['ddl']).replace("\n",'')
            logger.info(f"base sql:{ddl}")
            tableChanges = {}
            try:
                tableChanges = historyRecord['tableChanges'][0]
            except Exception as e:
                logger.error(e)
            if db == 'tt_member_215' or 'ADD INDEX' in ddl:
                continue
            if 'CHANGE' in ddl:
                sql_list = self.rename_column(db, table, ddl, tableChanges)
            elif 'CREATE TABLE' in ddl or 'create table' in ddl:
                sql_list = self.crate_table(db, table, ddl, tableChanges)
            elif 'DROP TABLE' in ddl:
                sql_list = [ddl]
            elif 'TRUNCATE TABLE' in ddl:
                sql_list = self.truncate_table(db, table)
            elif 'TRUNCATE ' in ddl:
                sql_list = self.truncate_table(db, table)
            elif 'ALTER TABLE' in ddl:
                sql_list = self.change_column(db, table, ddl, tableChanges)
            else:
                continue
            doris_db = MysqlHelper(db)
            for sql in sql_list:
                logger.info(sql)
                try:
                    # print("exe")
                    doris_db.execute_sql(sql)
                except:
                    self.error_do("base", "base", sql, "schema_change")

    def truncate_table(self, db, table):
        sql = []
        ddl = f"TRUNCATE TABLE {db}.{table}"
        sql.append(ddl)
        return sql

    def crate_table(self, db, table, ddl, tableChanges):
        column_p = r'create\s+table\s+`\S+`\s+like\s+`\S+`'

        sql = re.findall(column_p, ddl)

        if len(sql) > 0:
            return sql

        try:
            head = f'CREATE TABLE IF NOT EXISTS `{db}`.`{table}` ('
            table_info = tableChanges['table']
            columns = table_info['columns']
            keys = table_info['primaryKeyColumnNames']

            column_info = []
            for column_conf in columns:
                column_name = column_conf['name']
                typeName = column_conf['typeName']
                length = 0
                scale = 0
                if 'length' in column_conf.keys():
                    length = column_conf['length']
                    scale = column_conf.get('scale')

                column_type = self.format_column_type(length, scale, typeName)
                column_meta = f'`{column_name}` {column_type}'
                if column_name in keys:
                    index = keys.index(column_name)
                    column_info.insert(index, column_meta)
                else:
                    column_info.append(column_meta)
            end = f''' ) UNIQUE KEY({','.join(keys)})  DISTRIBUTED BY HASH({','.join(keys)}) BUCKETS 10  PROPERTIES ("replication_num" = "2","light_schema_change" = "true")'''
            desc_ = head + ','.join(column_info) + end
            sql.append(desc_)
        except :
            self.error_do(db, table, ddl, 'crate_table')
        return sql

    def get_column_name(self, column_info):
        p = r'`\S+`'
        res=[]
        columns_name = re.findall(p, column_info)[0].replace('`', '')
        # for column_name in columns_name:
        #     column_name=columns_name.replace('`', '')
        #     res.append(column_name)
        return columns_name

    def change_column(self, db, table, ddl, tableChanges):
        sql = []
        try:
            change_info = []
            update_info = []
            table_info = tableChanges['table']
            columns = table_info['columns']
            columns_info = {}
            for column in columns:
                columns_info[column['name']] = column
            pattern = r'ALTER TABLE\s+`\S+`\s+\S+|alter table\s+`\S+`\s+\S+'
            info = re.findall(pattern, ddl)[0]
            alter_type = info.split(' ')[-1]
            columns_base = ddl.replace(info, '')
            columns_base = columns_base.strip().split(f'{alter_type} ')
            if str(alter_type).upper() == 'DROP':
                for column in columns_base:
                    column_name = self.get_column_name(column)
                    change_info.append(f"DROP COLUMN {column_name}")
            else:
                for column in columns_base:
                    length = 0
                    scale = 0
                    column_name = self.get_column_name(column)
                    column_conf = columns_info[column_name]
                    typeName = column_conf['typeName']
                    if 'length' in column_conf.keys():
                        length = column_conf.get('length')
                        scale = column_conf.get('scale')
                    column_type = self.format_column_type(length, scale, typeName)
                    if str(alter_type).strip().upper() == 'MODIFY':
                        before_columns=self.get_columns(db,table)
                        before_column_type= before_columns.get(column_name.replace('`',''),None)
                        if before_column_type and column_type.upper().startswith(before_column_type) is False:
                            update=f"{column_name} = null"
                            update_info.append(update)
                    change_info.append(f"{alter_type} COLUMN {column_name} {column_type}")

            change_info_str = ','.join(change_info)

            column_type_change_sql = f'ALTER TABLE {db}.{table} {change_info_str}'
            if len(update_info)>0:
                update_info_str = ','.join(update_info)
                column_type_update_sql = f'UPDATE {db}.{table} SET {update_info_str}'
                sql.append(column_type_update_sql)
            sql.append(column_type_change_sql)

        except Exception as e:
            print(e)
            # self.error_do(db, table, ddl, 'change_column')
        return sql

    def rename_column(self, db, table, ddl, tableChanges):
        sql = []
        try:
            column_p1 = r'CHANGE\s+`\S+`\s+`\S+`\s+\S+'
            column_info = re.findall(column_p1, ddl)
            change_column = column_info[0]
            pro_list = change_column.split(' ')
            before_column = pro_list[1]
            after_column = pro_list[2]
            column_type = pro_list[3]
            table_info = tableChanges['table']
            columns_info = table_info['columns']
            typeName = 'int'
            length = 0
            scale = 0
            for column_conf in columns_info:
                if column_conf['name'] == after_column.replace('`', ''):
                    typeName = column_conf['typeName']

                    if 'length' in column_conf.keys():
                        length = column_conf.get('length')
                        scale = column_conf.get('scale')
                    break
            if before_column != after_column:
                column_rename = f'ALTER TABLE {db}.{table} RENAME COLUMN {before_column} {after_column}'
                sql.append(column_rename)
            column_type = self.format_column_type(length, scale, typeName)
            before_columns = self.get_columns(db, table)
            before_column_type= before_columns.get(after_column.replace('`',''),None)
            if before_column_type and column_type.upper().startswith(before_column_type) is False:
                update = f"{after_column} = null where {after_column} is not null"
                sql.append(f'UPDATE {db}.{table} SET {update}')
            column_type_change = f'ALTER TABLE {db}.{table} MODIFY COLUMN {after_column} {column_type}'
            sql.append(column_type_change)
        except:
            pass
            # self.error_do(db, table, ddl, 'rename_column')
        return sql

    def error_do(self, db, table, ddl, fnc):
        try:
            additional = {'db': db, 'table': table, 'ddl': ddl}
            timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
            error_info = {'project': 'schema_change', 'submodule': 'crate_table', 'additional': str(additional),
                          'timestamp': timestamp}
            doris_db = MysqlHelper(f"errors")
            error_sql_to = f"""insert into errors.errors(project,submodule,additional,timestamp)values ('schema_change',{fnc},'{additional}','{timestamp}')"""
            doris_db.execute_sql(error_sql_to)
            # self.producer.send(topic=self.error_topic, value=json.dumps(error_info).encode('utf-8'))
            self.error_msg.format(fnc, db, table, timestamp)
            send_email.sed("Doris SchemaChange Error", self.error_msg.format(fnc, db, table, timestamp))
        except:
            pass

    def format_column_type(self, length, scale, DATA_TYPE):
        COLUMN_TYPE = DATA_TYPE
        if DATA_TYPE == 'TEXT' or DATA_TYPE == 'LONGTEXT' or DATA_TYPE == 'TIMESTAMP' or DATA_TYPE == 'TINYTEXT' or DATA_TYPE == 'JSON':
            COLUMN_TYPE = 'STRING'
        elif DATA_TYPE == 'VARCHAR' or DATA_TYPE == 'CHAR':
            TYPE_LEN = length * 3
            COLUMN_TYPE = f'varchar({int(TYPE_LEN)})'
        elif DATA_TYPE == "VARCHAR":
            COLUMN_TYPE = "FLOAT"
        elif DATA_TYPE == "VARBINARY":
            COLUMN_TYPE = 'STRING'
        elif DATA_TYPE == "MEDIUMINT":
            COLUMN_TYPE = f'INT({length})'
        elif DATA_TYPE == "TIMESTAMP":
            COLUMN_TYPE = 'STRING'
        elif DATA_TYPE == "ENUM":
            COLUMN_TYPE = 'TINYINT'
        elif DATA_TYPE == "BIT":
            COLUMN_TYPE = 'TINYINT'
        elif DATA_TYPE == "SMALLINT":
            COLUMN_TYPE = 'SMALLINT'
        elif DATA_TYPE == "DECIMAL":
            COLUMN_TYPE = f'DECIMAL({length},{scale})'
        elif DATA_TYPE == "INT UNSIGNED":
            COLUMN_TYPE = 'INT'
        return COLUMN_TYPE
